package com.ksyun.alert.threadPool;

import com.ksyun.alert.calculate.CalculateImmediate;
import com.ksyun.alert.constant.RedisStreamConstant;
import com.ksyun.alert.dto.MetricDTO;
import com.ksyun.alert.utils.AnalyzeUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.stream.MapRecord;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 线程池类
 * @author Zhang
 * @date 2024-04-30
 */
@Configuration
@Slf4j
public class FixedThreadPool {

    /**
     * 线程池中的多线程从Redis中阻塞读取数据，得到数据后即可调用实时计算方法进行计算
     */

    @Resource
    private AnalyzeUtil analyzeUtil;

    @Resource
    private CalculateImmediate calculateImmediate;

    /**
     * 主机核心数目，计算密集型任务，线程池线程数目等于当前主机的核心线程数即可
     */
    public static final Integer CORES = Runtime.getRuntime().availableProcessors();

    @Bean(destroyMethod = "shutdown")
    public ExecutorService fixedGetRedisInfoThreadPool() {
        int numberOfThreads = CORES;
        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads, new DaemonThreadFactory());
        for (int i = 0; i < numberOfThreads; i++) {
            int id = i;
            AtomicBoolean flag = new AtomicBoolean(true);
            executorService.submit(() -> {
                while (true) {
                    //消费者名称
                    String consumerName = "consumer" + id;
                    //解决已读未确认的消息
                    if (flag.get()) {
                        analyzeUtil.handlePendingList(RedisStreamConstant.IMMEDIATE_GROUP, consumerName, RedisStreamConstant.STREAM_NAME);
                        flag.set(false);
                    }
                    //实时计算就拿一条记录就行
                    List<MapRecord<String, Object, Object>> records = analyzeUtil.readFromStream(1, consumerName);
                    //这一次读不到就下一轮阻塞读
                    if (records == null || records.isEmpty()) {
                        continue;
                    }
                    //第一条
                    MapRecord<String, Object, Object> entries = records.get(0);
                    //id用于xack确认
                    String recordId = entries.getId().toString();
                    List<List<MetricDTO>> lists = analyzeUtil.analyzeRecord(records);
                    try {
                        calculateImmediate.calculateNow(lists.get(0));
                    } catch (Exception e) {
                        log.error(e.getMessage());
                        //重试
                        flag.set(true);
                        continue;
                    }
                    //xack确认
                    analyzeUtil.redisStreamAckImmediate(recordId);
                }
            });
        }
        return executorService;
    }

    /**
     * 自定义线程工厂
     * 将线程标记为守护线程
     */
    static class DaemonThreadFactory implements ThreadFactory {
        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = Executors.defaultThreadFactory().newThread(runnable);
            thread.setDaemon(true);
            return thread;
        }
    }
}
